package com.lmf.cloud.springbootkafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Description kafka消费者各种模式：示例代码
 * todo 注意：(1) 是否配置自动提交offset模式、手动ack确认offset、单条消费或者批量消费模式；
 *  (2) 单条消费跟批量消费不同，都需要进行特殊配置且不能共存，否则代码会报错
 * 
 * @Author LiuMingFu
 * @Date 2024/8/11 10:58
 */
@Component
public class KafkaConsumer {


    /**
     * 单条消费、自动提交offset模式：获取一条消息，就会调用此方法一次
     *
     * @param message 消息内容
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "1", topics = "depart")
    public void receiveMessage1(String message) {
        //消费消息，处理业务逻辑
        System.out.println("KafkaConsumer.receiveMessage1.消费者接收到消息: " + message);
    }


    /**
     * 单条消费、自动提交offset模式：获取一条消息，就会调用此方法一次
     *
     * @param record 消息记录，里面包含一条消息的所有相关信息
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "2", topics = "depart", groupId = "xc-2")
    public void receiveMessage2(ConsumerRecord<String, String> record) {
        //消费消息，处理业务逻辑
        System.out.println("KafkaConsumer.receiveMessage2.消费者接收到消息: " + record);
    }


    /**
     * 单条消费、手动提交offset模式：获取一条消息，就会调用此方法一次
     *
     * @param message 单条消息内容，拉取的一批数据中的一条
     * @param ack     手动提交偏移量操作对象
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "3", topics = "depart", groupId = "xc-3")
    public void receiveMessage3(String message, Acknowledgment ack) {
        //消费消息，处理业务逻辑
        System.out.println("KafkaConsumer.receiveMessage3.消费者接收到消息: " + message);

        //手动提交偏移量，表示消费者已经消费了这条/批消息
        ack.acknowledge();
    }


    /**
     * 单条消费、手动提交offset模式：获取一条消息，就会调用此方法一次
     *
     * @param record 单条消息记录，拉取的一批数据中的一条
     * @param ack    手动提交偏移量操作对象
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "4", topics = "depart", groupId = "xc-4")
    public void receiveMessage4(ConsumerRecord<String, String> record, Acknowledgment ack) {
        //消费消息，处理业务逻辑
        System.out.println("KafkaConsumer.receiveMessage4.消费者接收到消息: " + record);

        //手动提交偏移量，表示消费者已经消费了这条/批消息
        ack.acknowledge();
    }


    /**
     * 批量消费、自动提交offset模式：获取一批消息，就会调用此方法一次
     *
     * @param recordList 消息记录集合，拉取的一批数据
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "5", topics = "depart", groupId = "xc-5")
    public void receiveMessage5(List<ConsumerRecord<String, String>> recordList) {
        //消费消息，处理业务逻辑
        for (ConsumerRecord<String, String> record : recordList) {
            System.out.println("KafkaConsumer.receiveMessage5.消费者接收到消息: " + record);
        }
    }


    /**
     * 批量消费、自动提交offset模式：获取一批消息，就会调用此方法一次
     *
     * @param messageList 消息记录集合，拉取的一批数据
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "6", topics = "depart", groupId = "xc-6")
    public void receiveMessage6(List<String> messageList) {
        //消费消息，处理业务逻辑
        for (String record : messageList) {
            System.out.println("KafkaConsumer.receiveMessage6.消费者接收到消息: " + record);
        }
    }


    /**
     * 批量消费、手动提交offset模式：获取一批消息，就会调用此方法一次
     *
     * @param recordList 消息记录集合，拉取的一批数据
     * @param ack        手动提交偏移量操作对象
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "7", topics = "depart", groupId = "xc-7")
    public void receiveMessage7(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        //消费消息，处理业务逻辑
        for (ConsumerRecord<String, String> record : recordList) {
            System.out.println("KafkaConsumer.receiveMessage7.消费者接收到消息: " + record);
        }
        //手动提交偏移量，表示消费者已经消费了这条/批消息
        ack.acknowledge();
    }


    /**
     * 批量消费、手动提交offset模式：获取一批消息，就会调用此方法一次
     *
     * @param messageList 消息内容集合，拉取的一批数据
     * @param ack         手动提交偏移量操作对象
     * @return
     * @author LiuMingFu
     * @date 2024/8/11
     */
    @KafkaListener(id = "8", topics = "depart", groupId = "xc-8")
    public void receiveMessage8(List<String> messageList, Acknowledgment ack) {
        //消费消息，处理业务逻辑
        for (String record : messageList) {
            System.out.println("KafkaConsumer.receiveMessage8.消费者接收到消息: " + record);
        }
        //手动提交偏移量，表示消费者已经消费了这条/批消息
        ack.acknowledge();
    }

}
